package com.nhncorp.pleiades.master;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.transaction.locking.ReadWriteLockManager;
import org.apache.commons.transaction.util.CommonsLoggingLogger;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Daemon;
import org.jgroups.Address;
import org.jgroups.ChannelException;
import org.jgroups.ChannelFactory;
import org.jgroups.ExtendedReceiverAdapter;
import org.jgroups.JChannel;
import org.jgroups.JChannelFactory;
import org.jgroups.MergeView;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.blocks.ReplicatedHashMap;

import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClient;
import com.nhncorp.pleiades.common.Constants;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.ens.EventMessage;
import com.nhncorp.pleiades.fs.PleiadesFileSystem;
import com.nhncorp.pleiades.hadoop.New_RPC;
import com.nhncorp.pleiades.hadoop.New_Server;
import com.nhncorp.pleiades.hadoop.WritableUtils;
import com.nhncorp.pleiades.msg.MasterServerInfo;
import com.nhncorp.pleiades.msg.MemberServerInfo;
import com.nhncorp.pleiades.msg.SplitServerInfo;
import com.nhncorp.pleiades.persistence.LockStatus;
import com.nhncorp.pleiades.protocol.AdminLockInfo;
import com.nhncorp.pleiades.protocol.ClientLockInfo;
import com.nhncorp.pleiades.protocol.ClientSessionInfo;
import com.nhncorp.pleiades.protocol.ClientSessionProtocol;
import com.nhncorp.pleiades.protocol.LockAdminProtocol;
import com.nhncorp.pleiades.protocol.LockMetaData;
import com.nhncorp.pleiades.protocol.LockServiceProtocol;
import com.nhncorp.pleiades.protocol.PleiadesInfoData;
import com.nhncorp.pleiades.util.NetworkUtil;
import com.nhncorp.pleiades.util.Sleeper;

@SuppressWarnings("unchecked")
public class ClusterManager implements ClusterManagerInterface {

  public static final Log LOG = LogFactory.getLog(ClusterManager.class.getName());
  private JChannel clusterChannel = null;
  private JChannel eventChannel = null;
  ReplicatedHashMap<String, Long> distributedSessionStore = null;
  ReplicatedHashMap<String, LockStatus> distributedLockStore = null;
  ConcurrentHashMap<String, Long> heartBeatMap = new ConcurrentHashMap<String, Long>();
  Executor sessionCullerThreadPool = null;
  private boolean isPrimaryNode = false;
  private boolean isReadyForEventService = false;
  protected final AtomicBoolean isPrimaryPartition = new AtomicBoolean(true);
  protected final AtomicBoolean stopRequested = new AtomicBoolean(false);
  private final Sleeper sleeper;
  private New_Server server;
  private PleiadesConfiguration conf;
  private Thread clusterChannelThread;
  private Thread heartBeatMonitorThread;
  private Thread heartBeatSynchronizer;
  private Thread lockServiceThread;
  private Thread eventChannelThread;
  private String clusterID;
  private PleiadesFileSystem fs;
  private String hostName;
  private String masterPleiades = "";
  private String prevMasterPleiades = "";
  private List<String> slavePleiadesList;
  private String masterEventManager;
  private List<String> slaveEventManagerList;
  private String dfsRoot = "";

  private Object splitSynchronizer;
  private List<LockStatus> invalidLockList;
  Queue<MemberServerInfo> slaveManagersQueue;
  AtomicInteger numberOfClusterManagers;
  private Controller controller = null;
  private List<LockEventSender> eventSenderList;
  private LinkedList<EventMessage> lockEventQueue;
  private LinkedList<Socket> socketQueue = new LinkedList<Socket>();
  private SplitBrainTimer splitBrainTimer = null;
  private Map<String, Long> splitInfoMap;
  private AtomicLong lastSplitTime;
  private FileTransactionManager fileTxnManager;
  private final ReadWriteLockManager memoryTxnManager = new ReadWriteLockManager(
      new CommonsLoggingLogger(LOG), 1000);
  private AtomicInteger fileTxnNumber;
  private AtomicInteger memoryTxnNumber;
  private AtomicInteger timeSinceLastGC;
  private Daemon heartbeatXceiveServer;

  static String getClusterManagerHostName(InetSocketAddress serverAddress) {
    return serverAddress.getHostName() + ":" + serverAddress.getPort();
  }

  static InetSocketAddress getClusterManagerAddress(PleiadesConfiguration conf)
      throws UnknownHostException {
    return NetworkUtil.getAddress(InetAddress.getLocalHost().getHostName() + ":"
        + conf.getInt("clusterManager.port", 8000));
  }

  public ClusterManager(String persistPropFile, PleiadesConfiguration pleiadesConf)
      throws Exception {
    super();

    System.setProperty("java.net.preferIPv4Stack", "true");
    LOG.debug("creating cluster manager...");
    if (pleiadesConf == null) {
      this.conf = new PleiadesConfiguration();
    } else {
      this.conf = pleiadesConf;
    }
    this.fs = PleiadesFileSystem.get(conf);
    this.clusterID = conf.get("clusterID");
    if (conf.getBoolean("pleiades.mode.splitBrain", false)) {
      LOG.info("pleiades starts with split-brain mode ...");
    }
    splitSynchronizer = new Object();
    invalidLockList = new ArrayList<LockStatus>();
    slaveManagersQueue = new ConcurrentLinkedQueue<MemberServerInfo>();
    numberOfClusterManagers = new AtomicInteger(0);
    this.sleeper = new Sleeper(1000, this.stopRequested);
    slavePleiadesList = new ArrayList<String>();
    slaveEventManagerList = new ArrayList<String>();
    splitBrainTimer = new SplitBrainTimer(10 * 1000);
    splitBrainTimer.start();
    splitInfoMap = new ConcurrentHashMap<String, Long>();
    lastSplitTime = new AtomicLong(0L);
    fileTxnNumber = new AtomicInteger(0);
    memoryTxnNumber = new AtomicInteger(0);
    fileTxnManager = new FileTransactionManager();
    timeSinceLastGC = new AtomicInteger(0);
    sessionCullerThreadPool = Executors.newFixedThreadPool(10);

    InetSocketAddress serverAddress = getClusterManagerAddress(conf);
    this.hostName = getClusterManagerHostName(serverAddress);
    LOG.info("====================cluster manager handler:" + serverAddress
        + "=====================");
    this.server = New_RPC.getServer(this, serverAddress.getHostName(), serverAddress.getPort(),
        conf.getInt("clusterManager.handler.count", 10), false, conf);

    ChannelFactory sessionFactory = new JChannelFactory();
    distributedSessionStore = new ReplicatedHashMap<String, Long>(clusterID
        + conf.get("clusterStorageName", Constants.CLUSTER_STORAGE_NAME), sessionFactory,
        Constants.STORAGE_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);
    distributedSessionStore.addNotifier(this);

    ChannelFactory lockFactory = new JChannelFactory();
    distributedLockStore = new ReplicatedHashMap<String, LockStatus>(clusterID
        + conf.get("distributedLockStore", Constants.LOCK_STORE_NAME), lockFactory,
        Constants.LOCK_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);
    LockServiceHandler lockServiceHandler = new LockServiceHandler();
    distributedLockStore.addNotifier(lockServiceHandler);

    clusterChannelThread = new Thread(new ClusterChannel(), "ClusterChannel Thread");
    eventChannelThread = new Thread(new EventChannel(), "EventChannel Thread");
    heartBeatMonitorThread = new Thread(new HeartBeatMonitor(), "HeartBeatMonitor Thread");
    lockServiceThread = new Thread(lockServiceHandler, "LockService Thread");
    heartBeatSynchronizer = new Thread(new HeartBeatSynchronizer(), "HeartBeatSynchronizer Thread");

    this.server.start();
    startHeartbeatReceiver();
    clusterChannelThread.start();
    eventChannelThread.start();
    heartBeatMonitorThread.start();
    heartBeatSynchronizer.start();
    lockServiceThread.start();
    distributedSessionStore.start(5 * 1000);
    distributedLockStore.start(5 * 1000);
  }

  private void startHeartbeatReceiver() {
    ServerSocket ss = null;
    int port = conf.getInt("clustermanager.heartbeat.port", 11000);
    String bindAddress = "0.0.0.0";
    try {
      ss = new ServerSocket(port, 0, InetAddress.getByName(bindAddress));
    } catch (IOException ie) {
      ie.printStackTrace();
      LOG.error("can not open heartbeat socket server at " + port + ", stop cluster manager", ie);
      System.exit(0);
    }
    this.heartbeatXceiveServer = new Daemon(new HeartbeatXceiveServer(ss));
    this.heartbeatXceiveServer.start();
  }

  class ClusterChannel extends ExtendedReceiverAdapter implements Runnable {
    private String localAddressForCluster = "";
    Object masterSynchronizer = new Object();

    public ClusterChannel() throws ChannelException {
      if (clusterChannel == null) {
        clusterChannel = new JChannel(Constants.CLUSTER_ADDRESS + Constants.SOCK_PROPERTY);
      }
      clusterChannel.setReceiver(this);
      clusterChannel.connect(clusterID
          + conf.get("clusterChannelName", Constants.CLUSTER_CHANNEL_NAME));
      controller = new ACLController(clusterChannel, conf);
    }

    public void run() {
      LOG.info("--------------- ClusterChannel thread started ---------------");
      try {
        clusterChannelLoop();
      } catch (IOException e) {
        LOG.error("Error in ClusterChannel thread", e);
      }
    }

    private void clusterChannelLoop() throws IOException {
      long lastUpdatedTime = System.currentTimeMillis();
      long interval = conf.getLong("clusterManager.acl.reloadInterval", 10 * 1000);
      int maxTxnNumber = 200000000;
      int initTxnNumber = 0;
      int gcInterval = 100000;

      while (!stopRequested.get()) {
        try {
          if (isPrimaryNode) {
            if ((System.currentTimeMillis() - lastUpdatedTime) > interval) {
              controller.updateServiceTable();
              lastUpdatedTime = System.currentTimeMillis();
            }
            if (fileTxnNumber.get() > maxTxnNumber) {
              fileTxnNumber.set(initTxnNumber);
            }
            if (memoryTxnNumber.get() > maxTxnNumber) {
              memoryTxnNumber.set(initTxnNumber);
            }
            if (timeSinceLastGC.get() > gcInterval) {
              System.gc();
              timeSinceLastGC.set(0);
              LOG.info("GC Interval " + gcInterval + " has been reached and GC is executed");
            }
          }
          Thread.sleep(100);
        } catch (InterruptedException e) {
          if (clusterChannel != null || clusterChannel.isOpen()) {
            clusterChannel.close();
          }
          break;
        }
      }
      clusterChannel.close();
    }

    public void receive(Message msg) {
      Object message = msg.getObject();
      try {
        if (isPrimaryNode) {
          if (message instanceof MemberServerInfo) {
            slaveManagersQueue.add((MemberServerInfo) message);
          } else if (message instanceof SplitServerInfo) {
            LOG.debug("received SplitServerInfo : " + ((SplitServerInfo) message));
            splitInfoMap.put(((SplitServerInfo) message).getServerAddress(),
                ((SplitServerInfo) message).getPartitionTime());
          }
        } else {
          if (message instanceof ServiceACL[]) {
            controller.storeServiceTable((ServiceACL[]) message);
          } else if (message instanceof SplitServerInfo) {
            LOG.debug("received SplitServerInfo : " + ((SplitServerInfo) message));
            splitInfoMap.put(((SplitServerInfo) message).getServerAddress(),
                ((SplitServerInfo) message).getPartitionTime());
          } else if (message instanceof MasterServerInfo) {
            synchronized (masterSynchronizer) {
              masterPleiades = ((MasterServerInfo) message).getServerAddress();
            }
          }
        }
      } catch (Exception e) {
        LOG.error("Error while receiving message", e);
      }
    }

    public void viewAccepted(View view) {
      String creator = view.getCreator().toString();
      LOG.debug("view creator : -----" + creator + "-----");
      localAddressForCluster = clusterChannel.getLocalAddress().toString();
      if (localAddressForCluster.equals("")) {
        while (true) {
          if (clusterChannel.isOpen()) {
            localAddressForCluster = clusterChannel.getLocalAddress().toString();
            if (!localAddressForCluster.equals("")) {
              break;
            }
          }
          sleeper.sleepMilliSeconds(10);
        }
      }

      if (view instanceof MergeView) {
        if (conf.getBoolean("pleiades.mode.splitBrain", false)) {
          LOG.info("merge view : " + view);
          isPrimaryPartition.set(false);
          splitBrainTimer.stop();
          SplitBrainHandler handler = new SplitBrainHandler((MergeView) view);
          handler.start();
        }
      } else {
        if (conf.getBoolean("pleiades.mode.splitBrain", false)) {
          if (view.size() >= conf.getInt("pleiades.cell.majorityCount", 3)) {
            isPrimaryPartition.set(true);
            splitBrainTimer.start();
            LOG.info("become primary partition " + view.getCreator());
          } else {
            isPrimaryPartition.set(false);
            splitBrainTimer.stop();
            LOG.info("not yet primary partition " + view.getCreator());
          }
        }

        if (localAddressForCluster.equals(creator)) {
          isPrimaryNode = true;
          masterPleiades = getHostName();
          LOG.info("master pleiades : " + masterPleiades);
          LOG.info("<<<<<primary : " + localAddressForCluster + " for cluster channel>>>>>");
          LOG.info("primary node's view : " + view);
        } else {
          isPrimaryNode = false;
          synchronized (masterSynchronizer) {
            if (masterPleiades.equals("")) {
              // need to fix : sending request to get master pleiades info
              String tempMaster = creator.substring(0, creator.indexOf(":"));
              masterPleiades = tempMaster + hostName.substring(hostName.indexOf(":"));
            }
          }
          LOG.info("<<<<<secondary : " + localAddressForCluster + " for cluster channel>>>>>");
        }

        new Thread() {
          public void run() {
            try {
              if (isPrimaryNode) {
                try {
                  updateSlaveManagerList();
                  sleeper.sleepMilliSeconds(500);
                  controller.spreadServiceTable();
                } catch (IOException e) {
                  e.printStackTrace();
                }
              }
              while (!clusterChannel.isConnected()) {
                sleeper.sleepMilliSeconds(10);
              }
              if (isPrimaryNode) {
                if (!masterPleiades.equals(prevMasterPleiades)) {
                  clusterChannel.send(new Message(null, null, new MasterServerInfo(getHostName())));
                  sendLockEventToManager(new EventMessage(getHostName(),
                      LockServiceClient.Events.MASTER_FAILOVER.ordinal(), "", false, System
                          .currentTimeMillis()));
                  prevMasterPleiades = masterPleiades;
                }
              } else {
                MemberServerInfo mServerInfo = new MemberServerInfo(getHostName());
                mServerInfo.setMaster(false);
                clusterChannel.send(new Message(clusterChannel.getView().getCreator(), null,
                    mServerInfo));
              }
            } catch (ChannelException e) {
              e.printStackTrace();
            }
          }
        }.start();

      }
    }

    private void updateSlaveManagerList() {
      new Thread() {
        public void run() {
          MemberServerInfo[] copiedGroupMembers;
          sleeper.sleepMilliSeconds(500);
          numberOfClusterManagers.set(clusterChannel.getView().size());
          synchronized (slaveManagersQueue) {
            while (slaveManagersQueue.size() > numberOfClusterManagers.get()) {
              slaveManagersQueue.remove();
            }
            MemberServerInfo mServerInfo = new MemberServerInfo(getHostName());
            mServerInfo.setMaster(true);
            slaveManagersQueue.add(mServerInfo);
            copiedGroupMembers = new MemberServerInfo[slaveManagersQueue.size()];
            slaveManagersQueue.toArray(copiedGroupMembers);
            boolean masterClusterManagerIncluded = false;
            Set<String> memberSet = new HashSet<String>();
            for (MemberServerInfo member : copiedGroupMembers) {
              if (member.isMaster()) {
                if (member.getServerAddress().equals(hostName)) {
                  if (masterClusterManagerIncluded) {
                    slaveManagersQueue.remove(member);
                  } else {
                    masterClusterManagerIncluded = true;
                    memberSet.add(hostName);
                  }
                } else {
                  slaveManagersQueue.remove(member);
                }
              } else {
                if (memberSet.contains(member.getServerAddress())) {
                  slaveManagersQueue.remove(member);
                } else {
                  memberSet.add(member.getServerAddress());
                }
              }
            }
            slavePleiadesList.clear();
            for (MemberServerInfo manager : slaveManagersQueue) {
              if (!manager.isMaster()) {
                slavePleiadesList.add(manager.getServerAddress());
                LOG.info("slave pleiades : " + manager.getServerAddress());
              }
            }
            if (slavePleiadesList.isEmpty()) {
              slavePleiadesList.add(getHostName());
            }
          }
        }
      }.start();
    }
  }

  private class SplitBrainHandler extends Thread {
    AtomicBoolean splitBrainMaster;
    MergeView view;

    private SplitBrainHandler(MergeView view) {
      splitBrainMaster = new AtomicBoolean(false);
      this.view = view;
    }

    public void run() {
      synchronized (splitSynchronizer) {
        if ((System.currentTimeMillis() - lastSplitTime.get()) < 10 * 1000) {
          return;
        }
        handleSplitBrain();
      }
    }

    private void handleSplitBrain() {

      LOG.info("split brain handler starts ...");
      List<View> primaryViews = new ArrayList<View>();
      List<View> secondaryViews = new ArrayList<View>();
      Vector<View> subGroups = view.getSubgroups();
      for (View entry : subGroups) {
        if (entry.size() >= conf.getInt("pleiades.cell.majorityCount", 3)) {
          primaryViews.add((View) entry.clone());
        } else if (entry.size() == (conf.getInt("pleiades.cell.majorityCount", 3) - 1)) {
          secondaryViews.add((View) entry.clone());
        }
      }
      View master_view = null;
      Address local_addr = clusterChannel.getLocalAddress();
      if (primaryViews.isEmpty()) {
        LOG.fatal("fatal split brain syndrome : no majority sub group!!!");
        // Send SMS Alarm to the Administrator if possible
        if (conf.getBoolean("pleiades.split.enableBestEffort", false)) {
          if (secondaryViews.isEmpty()) {
            LOG.fatal("network harmagedon situation !!!");
          } else {
            LOG.fatal("try to select split master at best effort manner");
            master_view = secondaryViews.get(0);
          }
        }
      } else {
        int numOfPrimaryViews = primaryViews.size();
        for (View entry : primaryViews) {
          LOG.info("cordinator of subgroup : " + entry.getCreator());
          if (local_addr.toString().equals(entry.getCreator().toString())) {
            try {
              long partitionTime = splitBrainTimer.stayTime();
              splitInfoMap.put(local_addr.toString(), partitionTime);
              SplitServerInfo splitServerInfo = new SplitServerInfo(local_addr.toString());
              splitServerInfo.setPartitionTime(partitionTime);
              clusterChannel.send(new Message(null, null, splitServerInfo));
            } catch (Exception e) {
              e.printStackTrace();
            }
          }
        }
        int maxCount = 0;
        int numberOfRetry = conf.getInt("pleiades.split.propagation.retry", 10);
        boolean receivedSplitInfo = true;
        while (splitInfoMap.size() < numOfPrimaryViews) {
          if (maxCount > numberOfRetry) {
            LOG.fatal("failed to get all splitBrainInfo and got only " + splitInfoMap.size());
            receivedSplitInfo = false;
            break;
          }
          sleeper.sleep();
          ++maxCount;
        }
        if (receivedSplitInfo) {
          String tempPrimary = "";
          long tempPartitionTime = 0;
          for (Map.Entry<String, Long> entry : splitInfoMap.entrySet()) {
            LOG.info("split info map : " + entry.getKey() + "," + entry.getValue());
            if (entry.getValue() > tempPartitionTime) {
              tempPrimary = entry.getKey();
              tempPartitionTime = entry.getValue();
            }
          }
          for (View entry : primaryViews) {
            if (tempPrimary.equals(entry.getCreator().toString())) {
              master_view = entry;
              LOG.info(entry.getCreator() + " is elected as a master view");
              break;
            }
          }
        } else {
          master_view = primaryViews.get(0);
        }
      }

      if (master_view != null) {
        try {
          LOG.info("View ID : " + master_view.getVid());
          if (local_addr.toString().equals(master_view.getCreator().toString())) {
            splitBrainMaster.set(true);
          } else {
            clusterChannel.close();
            sleeper.sleep();
            clusterChannel = new JChannel(Constants.CLUSTER_ADDRESS + Constants.SOCK_PROPERTY);
            clusterChannel.connect(clusterID
                + conf.get("clusterChannelName", Constants.CLUSTER_CHANNEL_NAME));
            isPrimaryNode = false;
          }
          if (!master_view.getMembers().contains(local_addr)) {
            LOG.info("I (" + local_addr + ") am not member of the primary partition ("
                + master_view + "), will re-acquire the state");

            distributedSessionStore._clear();
            distributedLockStore._clear();
            distributedSessionStore.stop();
            distributedLockStore.stop();
            sleeper.sleep();
            distributedSessionStore = new ReplicatedHashMap<String, Long>(clusterID
                + conf.get("clusterStorageName", Constants.CLUSTER_STORAGE_NAME), null,
                Constants.STORAGE_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);
            distributedLockStore = new ReplicatedHashMap<String, LockStatus>(clusterID
                + conf.get("distributedLockStore", Constants.LOCK_STORE_NAME), null,
                Constants.LOCK_ADDRESS + Constants.SOCK_PROPERTY, true, 1000);

          } else {
            LOG.info("I (" + local_addr + ") am member of the primary partition (" + master_view
                + ")," + " will do nothing");
          }
        } catch (Exception e) {
          e.printStackTrace();
        } finally {
          isPrimaryPartition.set(true);
          splitInfoMap.clear();
          splitBrainTimer.start();
          lastSplitTime.set(System.currentTimeMillis());
          dumpLockStatus();
        }
      }
    }
  }

  class HeartBeatSynchronizer implements Runnable {
    static final int SYNCHRONIZATION_GAP = 5000;

    long lastSyncTime;

    public void run() {
      try {
        Thread.sleep(Math.abs(new Random().nextInt()) % 10000);
      } catch (InterruptedException e) {
        return;
      }

      lastSyncTime = System.currentTimeMillis();
      LOG.info("--------------- HeartbeatSynchronizer thread started ---------------");
      while (true) {
        try {
          Thread.sleep(SYNCHRONIZATION_GAP);  
          if (!heartBeatMap.isEmpty()
              && System.currentTimeMillis() - lastSyncTime > SYNCHRONIZATION_GAP) {
            distributedSessionStore.putAll(heartBeatMap);
            heartBeatMap.clear();
            lastSyncTime = System.currentTimeMillis();
          }
        } catch (InterruptedException e) {
          LOG.info("HeartBeatSynchronizer ends");
          break;
        } catch (Exception e) {
          LOG.warn("Error in synchronizing heartbeat info, continue", e);
        }
      }
    }
  }

  class HeartBeatMonitor implements Runnable {
    private long heartbeatTimeout = conf.getLong("clusterManager.heartbeat.timeout", 10 * 1000);;
    private ArrayList<String> pendingSessionList = new ArrayList<String>();

    public void run() {
      LOG.info("--------------- HeartbeatMonitor thread started ---------------");
      monitorLoop();
    }

    private void monitorLoop() {
      while (!stopRequested.get()) {
        try {
          if (isPrimaryNode && isPrimaryPartition.get()) {
            if (!pendingSessionList.isEmpty()) {
              createExpiredSessionCullerThread().start();
            }

            if (heartBeatMap.isEmpty() && distributedSessionStore.isEmpty()) {
              sleeper.sleep();
              continue;
            }

            long currentTime = System.currentTimeMillis();
            releaseExpiredSessionsIn(heartBeatMap, currentTime);
            releaseExpiredSessionsIn(distributedSessionStore, heartBeatMap, currentTime);
          }

          sleeper.sleep();

        } catch (Exception e) {
          LOG.warn("Error in monitoring heartbeat", e);
        }
      }
    }

    private Thread createExpiredSessionCullerThread() {
      return new Thread() {
        List<String> clonedList;

        public void run() {
          synchronized (pendingSessionList) {
            clonedList = (List<String>) pendingSessionList.clone();
            pendingSessionList.clear();
          }

          for (String entry : clonedList) {
            if (heartBeatMap.containsKey(entry) || distributedSessionStore.containsKey(entry)) {
              LOG.info("session expired host " + entry + " is now alive");
              clonedList.remove(entry);
            } else {
              if (releaseAllLocks(entry)) {
                clonedList.remove(entry);
              } else {
                LOG.warn("releaseAllLocks for expiredHostList entry " + entry + " failed");
                synchronized (pendingSessionList) {
                  pendingSessionList.add(entry);
                }
              }
            }
          }
        }
      };
    }

    void releaseExpiredSessionsIn(final Map<String, Long> heartbeatMap, long currentTime) {
      releaseExpiredSessionsIn(heartbeatMap, null, currentTime);
    }

    void releaseExpiredSessionsIn(final Map<String, Long> heartbeatMap, Map<String, Long> alterMap,
        long currentTime) {
      for (Map.Entry<String, Long> entry : heartbeatMap.entrySet()) {
        final String creator = entry.getKey();

        long serviceTimeout = getServiceTimeout(creator);
        if ((currentTime - entry.getValue()) > serviceTimeout) {

          if (alterMap != null) {
            Long lastActivityTime = null;
            if ((lastActivityTime = alterMap.get(creator)) != null) {
              if ((currentTime - lastActivityTime) <= serviceTimeout) {
                heartBeatMap.put(creator, lastActivityTime);
                continue;
              }
            }
          }

          LOG.info("client : " + creator + " 'session is expired and release the locks");

          executeToSessionCullerThreadPool(heartbeatMap, creator);
        }
      }
    }

    private void executeToSessionCullerThreadPool(final Map<String, Long> heartbeatMap,
        final String creator) {
      sessionCullerThreadPool.execute(new Runnable() {
        public void run() {
          heartbeatMap.remove(creator);

          if (!releaseAllLocks(creator, true)) {
            LOG.debug("releaseAllLocks failed " + creator + " will remove later");
            synchronized (pendingSessionList) {
              pendingSessionList.add(creator);
            }
          }
        }
      });
    }

    private boolean isLockOwnerLive(String ownerOfLocks) {
      try {
        StringTokenizer st = new StringTokenizer(ownerOfLocks, "||");
        if (st.countTokens() != 3) {
          return false;
        }
        String ownerServiceId = st.nextToken();
        String ownerHostName = st.nextToken();
        int ownerHeartbeatPort = Integer.parseInt(st.nextToken());

        if (ownerHeartbeatPort == 0) {
          return false;
        }

        String onlyHostName = ownerHostName;
        if (ownerHostName.indexOf(":") > 0) {
          onlyHostName = ownerHostName.substring(0, ownerHostName.indexOf(":"));
        }

        Socket socket = null;
        BufferedInputStream in = null;
        try {
          socket = new Socket();
          socket.setSoLinger(true, 1);
          socket.connect(NetworkUtil.getAddress(onlyHostName + ":" + ownerHeartbeatPort), 3 * 1000);
          in = new BufferedInputStream(socket.getInputStream());
          byte[] buf = new byte[10];
          int readCount = in.read(buf);

          return true;
        } finally {
          if (in != null) {
            try {
              in.close();
            } catch (Exception e) {
            }
          }
          if (socket != null) {
            try {
              socket.close();
            } catch (Exception e) {
            }
          }
        }
      } catch (Exception e) {
        LOG.debug("Error while connect to client:" + ownerOfLocks + ":" + e.getMessage());
        return false;
      }
    }

    private long getServiceTimeout(String creator) {
      return creator.indexOf("||") > 0 ? conf.getLong(creator.substring(0, creator.indexOf("||"))
          + ".heartbeat.timeout", heartbeatTimeout) : this.heartbeatTimeout;
    }
  }

  class HeartbeatXceiveServer implements Runnable {
    boolean shouldListen = true;
    ServerSocket ss;
    List<HeartbeatProcessor> hbProcessorList;
    int processorCount = 10;

    public HeartbeatXceiveServer(ServerSocket ss) {
      this.ss = ss;
      processorCount = conf.getInt("clusterManager.heartbeat.handlerCount", 10);
      hbProcessorList = new ArrayList<HeartbeatProcessor>();
      for (int count = 0; count < processorCount; ++count) {
        hbProcessorList.add(new HeartbeatProcessor("Heartbeat Processor Thread "
            + String.valueOf(count)));
      }
    }

    public void run() {
      try {
        LOG.info("--------------- HeartbeatXceiveServer started ---------------");
        for (HeartbeatProcessor processor : hbProcessorList) {
          processor.start();
        }
        while (shouldListen) {
          Socket s = ss.accept();
          s.setSoLinger(true, 1);
          synchronized (socketQueue) {
            socketQueue.add(s);
            socketQueue.notifyAll();
          }
        }
        ss.close();
      } catch (IOException ie) {
        LOG.info("exiting cluster manager HeartbeatXceiveServer due to " + ie.toString());
      }
    }

    public void kill() {
      this.shouldListen = false;
      try {
        this.ss.close();
      } catch (IOException iex) {
      }
    }
  }

  class HeartbeatProcessor extends Thread {
    DataInputStream in = null;
    Socket socket = null;

    HeartbeatProcessor(String name) {
      super(name);
    }

    public void run() {
      while (!stopRequested.get()) {
        try {
          synchronized (socketQueue) {
            while (socketQueue.isEmpty()) {
              socketQueue.wait();
            }
            socket = socketQueue.removeFirst();
          }
          in = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
          String ownerId = WritableUtils.readString(in);
          LOG.debug("heartbeat " + ownerId);
          if(!ownerId.equals("R")) {
            updateHeartbeat(ownerId);  
          }
        } catch (InterruptedException e) {
          return;
        } catch (Exception e) {
          e.printStackTrace();
          LOG.error("Heartbeat process error", e);
        } finally {
          if (in != null) {
            try {
              in.close();
            } catch (Exception e) {
            }
          }
          try {
            socket.close();
            socket = null;
          } catch (Exception e) {
          }
        }
      }
    }
  };

  class EventChannel extends ReceiverAdapter implements Runnable {
    private String localAddressForEvent = "";

    public EventChannel() throws ChannelException {
    }

    public void run() {
      try {
        LOG.info("--------------- EventChannel thread started ---------------");
        boolean exitOnlyWhenPrimary = false;
        while (!exitOnlyWhenPrimary) {
          if (isPrimaryNode) {
            exitOnlyWhenPrimary = true;
          }
          sleeper.sleep();
        }
        eventChannelLoop();

      } catch (ChannelException e) {
        LOG.error("EventChannel thread error", e);
      }
    }

    private void eventChannelLoop() throws ChannelException {
      String eventCreator = "";
      while (!stopRequested.get()) {
        try {
          if (eventChannel == null) {
            eventChannel = new JChannel(Constants.EVENT_ADDRESS + Constants.CLIENT_PROPERTY);
            eventChannel.setReceiver(this);
            eventChannel.connect(clusterID
                + conf.get("eventChannelName", Constants.EVENT_CHANNEL_NAME));
          } else if (!eventChannel.isConnected()) {
            eventChannel.connect(clusterID
                + conf.get("eventChannelName", Constants.EVENT_CHANNEL_NAME));
          }
          View view = eventChannel.getView();
          eventCreator = view.getCreator().toString();
          localAddressForEvent = eventChannel.getLocalAddress().toString();

          if (eventCreator.equals(localAddressForEvent)) {
            isReadyForEventService = false;
            sleeper.sleep();
            eventChannel.close();
            eventChannel = null;
          } else {
            isReadyForEventService = true;
          }
          sleeper.sleep();
        } catch (ChannelException e) {
          return;
        }
      }
      if (eventChannel != null || eventChannel.isConnected()) {
        eventChannel.close();
      }
    }

    public void receive(Message msg) {

      if (msg.getObject() instanceof MemberServerInfo[]) {
        MemberServerInfo[] membershipInfos = (MemberServerInfo[]) msg.getObject();
        synchronized (slaveEventManagerList) {
          slaveEventManagerList.clear();
          if (membershipInfos.length == 1) {
            masterEventManager = membershipInfos[0].getServerAddress();
            LOG.info("master event manager : " + masterEventManager);
            return;
          }
          for (MemberServerInfo membership : membershipInfos) {
            if (membership.isMaster()) {
              masterEventManager = membership.getServerAddress();
              LOG.info("master event manager : " + masterEventManager);
            } else {
              slaveEventManagerList.add(membership.getServerAddress());
              LOG.info("slave event manager: " + membership.getServerAddress());
            }
          }
        }
      }
    }
  }

  @SuppressWarnings("unchecked")
  /*
   * At this moment, this class is only used to dispose invalid locks, sooner or
   * later, a variety of lock handling mechanism could be implemented with this
   */
  class LockServiceHandler implements Runnable, ReplicatedHashMap.Notification {
    public LockServiceHandler() {
    }

    public void run() {
      LOG.info("--------------- LockService thread started ---------------");
      loopForLockHandler();
    }

    private void loopForLockHandler() {
      while (!stopRequested.get()) {
        try {
          if (isPrimaryNode) {
            if (invalidLockList != null && invalidLockList.size() > 0) {
              for (LockStatus entry : invalidLockList) {
                LockStatus removedLock = distributedLockStore.remove(entry.getLockId());
                invalidLockList.remove(removedLock);
                removedLock = null;
              }
            }
            if(!invalidLockList.isEmpty()) {
              LOG.info("still invalid locks " + String.valueOf(invalidLockList.size()) + "ea exist");
            }
          }
          sleeper.sleep();
        } catch (Exception e) {
          LOG.error("loopForLockHandler error", e);
          return;
        }
      }
    }

    public void contentsCleared() {
    }

    public void contentsSet(Map map) {
    }

    public void viewChange(View view, Vector joined, Vector existed) {
    }

    public void entryRemoved(Serializable arg0) {
    }

    public void entrySet(Serializable key, Serializable value) {
    }
  }

  public void contentsCleared() {
  }

  public void contentsSet(Map map) {
  }

  public void viewChange(View view, Vector joined, Vector existed) {
  }

  public void entryRemoved(Serializable key) {
    LOG.debug(key.toString() + " is removed from clusterChannel()");
  }

  public void entrySet(Serializable key, Serializable value) {
    LOG.debug(key.toString() + " is added from clusterChannel()");
  }

  // ///////////////////////////////////////////////////
  // LockServiceProtocol
  // ///////////////////////////////////////////////////

  public boolean isLockManagerReady() {
    if (isPrimaryNode && isPrimaryPartition.get())
      return true;
    else {
      LOG.fatal("primary node : " + isPrimaryNode + ", " + "primary partition : "
          + isPrimaryPartition.get());
      return false;
    }
  }

  public boolean tryLock(ClientLockInfo lockInfo) {
    LOG.debug("!!!TRYLOCK!!! : " + lockInfo);
    updateHeartbeat(lockInfo.getCreator());

    boolean isLocked = false;
    try {
      if (isPrimaryNode && isPrimaryPartition.get()) {
        if (!controller.isValidateLock(lockInfo.getLockId())) {
          LOG.info("invalid lock path: " + lockInfo.getLockId());
          return isLocked;
        }
        LockStatus writeLockStatus = distributedLockStore.get(lockInfo.getLockId());
        if (writeLockStatus != null && !writeLockStatus.getHostName().equals(lockInfo.getCreator())) {
          return isLocked;
        }
        isLocked = true;
      }
    } catch (Exception e) {
      LOG.warn("Error while trying lock", e);

    } finally {
    }
    return isLocked;
  }

  public String lock(ClientLockInfo lockInfo) {
    updateHeartbeat(lockInfo.getCreator());

    if (!isPrimaryNode) {
      return getMasterPleiades();
    }
    boolean isLocked = false;
    Integer memTxn = memoryTxnNumber.getAndIncrement();
    try {
      LOG.debug("!!!LOCK!!! : " + lockInfo);
      if (isPrimaryPartition.get()) {
        memoryTxnManager.writeLock(memTxn, lockInfo.getLockId());
        LockStatus writeLockStatus = distributedLockStore.get(lockInfo.getLockId());

        if (writeLockStatus != null && !writeLockStatus.getHostName().equals(lockInfo.getCreator())) {
          LOG.debug("lock " + lockInfo + " distrbuted lock owner exists : " + writeLockStatus);
        } else {
          if (lockInfo.getLockMode() == LockServiceClient.LockMode.LOCK_SHARED.ordinal()) {
            distributedLockStore.put(lockInfo.getLockId() + "::" + lockInfo.getCreator(),
                new LockStatus(lockInfo.getCreator(), lockInfo.getLockId(),
                    LockServiceClient.LockMode.LOCK_SHARED.ordinal()));
          } else {
            LockStatus lockStatus = new LockStatus(lockInfo.getCreator(), lockInfo.getLockId(),
                LockServiceClient.LockMode.LOCK_EXCLUSIVE.ordinal(), lockInfo.getMetaData());
            distributedLockStore.put(lockInfo.getLockId(), lockStatus);

            /** this is temporary...need to fix and tune later * */
            int exitCount = 0;
            int waitTime = 0;

            while (distributedLockStore.get(lockInfo.getLockId()) == null) {
              LOG.debug("wait until lockInfo is propagated : " + lockInfo);
              sleeper.sleepMilliSeconds(++waitTime);
              exitCount++;
              if (exitCount > 10) {
                if (distributedLockStore.remove(lockInfo.getLockId(), lockStatus)) {
                } else {
                  invalidLockList.add(lockStatus);
                }
                throw new IOException("lockInfo propagation failed and return false : " + lockInfo);
              }
            }
          }
          isLocked = true;
        }
      }
    } catch (Exception e) {
      isLocked = false;
      LOG.error("error while locking", e);

    } finally {
      memoryTxnManager.release(memTxn, lockInfo.getLockId());
    }
    return String.valueOf(isLocked);
  }

  public String unlock(ClientLockInfo lockInfo) {
    updateHeartbeat(lockInfo.getCreator());

    if (!isPrimaryNode) {
      return getMasterPleiades();
    }
    boolean isUnLocked = false;
    Integer memTxn = memoryTxnNumber.getAndIncrement();
    try {
      LOG.debug("!!!UNLOCK!!! " + lockInfo);
      if (isPrimaryPartition.get()) {

        memoryTxnManager.writeLock(memTxn, lockInfo.getLockId());
        LockStatus writeLockStatus = distributedLockStore.get(lockInfo.getLockId());

        if (writeLockStatus != null && !writeLockStatus.getHostName().equals(lockInfo.getCreator())) {
          LOG.debug("unlock " + lockInfo + " distrbuted lock owner exists : " + writeLockStatus);

        } else if (lockInfo.getLockMode() == LockServiceClient.LockMode.LOCK_SHARED.ordinal()) {
          distributedLockStore.remove(lockInfo.getLockId() + "::" + lockInfo.getCreator());
          isUnLocked = true;
        } else {
          distributedLockStore.remove(lockInfo.getLockId());
          isUnLocked = true;
        }
      }
    } catch (Exception e) {
      isUnLocked = false;
      LOG.error("error while unlocking", e);
    } finally {
      memoryTxnManager.release(memTxn, lockInfo.getLockId());
    }
    return String.valueOf(isUnLocked);

  }

  public boolean lockExist(String path) throws IOException {
    try {
      if (distributedLockStore.containsKey(path)) {
        return true;
      }
    } catch (Exception e) {
      LOG.error("error while checking lock existence", e);
      throw new IOException(e.getMessage());
    }
    return false;
  }

  public LockStatus[] listLock(String path) {

    boolean isSuccess = false;
    int maxNumberOfEntry = 1000;
    List<LockStatus> lockList = new ArrayList<LockStatus>();

    try {
      if (path.endsWith("/")) {
        path = path.substring(0, (path.length() - 1));
      }
      int childIndex = 0;
      for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
        if (entry.getKey().startsWith(path) && !entry.getKey().equals(path)) {
          childIndex = entry.getKey().indexOf("/", path.length());
          if (childIndex > 0) {
            if (entry.getKey().indexOf("/", childIndex + 1) < 0) {
              if (!entry.getKey().endsWith("/")) {
                lockList.add(entry.getValue());
                if (lockList.size() > maxNumberOfEntry) {
                  lockList.clear();
                  lockList = null;
                  throw new IOException("too many lock status information");
                }
              }
            }
          }
        }
      }
      isSuccess = true;
    } catch (Exception e) {
      isSuccess = false;
      LOG.error("error while listing lock path[" + path + "]", e);
    } finally {
      if (!isSuccess) {
        if (lockList != null && !lockList.isEmpty()) {
          lockList.clear();
          lockList = null;
        }
        return null;
      }
    }

    return lockList.toArray(new LockStatus[] {});
  }

  public boolean releaseAllLocks(String owner) {
    return releaseAllLocks(owner, false);
  }

  private boolean releaseAllLocks(String owner, boolean isClientFailed) {
    boolean isSuccess = false;
    if (!isPrimaryNode || !isPrimaryPartition.get()) {
      LOG.fatal("not primary node or partition");
      return isSuccess;
    }

    Set<String> nodes = new TreeSet<String>();
    long generationNumber = 0;
    try {
      for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
        if (entry.getValue().getHostName().equals(owner)) {
          LOG.debug("removing lock :" + entry.getKey());
          if (generationNumber == 0) {
            generationNumber = System.nanoTime();
          }
          if (distributedLockStore.remove(entry.getValue().getLockId()) == null) {
            if (distributedLockStore.containsKey(entry.getValue().getLockId())) {
              LOG.warn("removing distributed lock failed :" + entry.getKey());
            }
          }
          if (isClientFailed) {
            nodes.add(entry.getValue().getLockId());
          }
        }
      }
      isSuccess = true;
    } finally {

      if (isClientFailed) {
        try {
          sendLockEventToManager(new EventMessage(owner, LockServiceClient.Events.CLIENT_EXPIRED
              .ordinal(), "", false, generationNumber));
          Iterator<String> iter = nodes.iterator();
          while (iter.hasNext()) {
            sendLockEventToManager(new EventMessage(owner, LockServiceClient.Events.LOCK_EXPIRED
                .ordinal(), iter.next(), false, generationNumber));
          }
        } catch (ChannelException e) {
          LOG.error("error while releasing all locks, owner[" + owner + "]", e);
        }
      }
    }

    return isSuccess;
  }

  private void sendLockEventToManager(EventMessage eventMessage) throws ChannelException {
    boolean isMessageSent = false;
    int loopCount = 0;
    while (!isMessageSent) {
      if (loopCount > 10) {
        LOG.debug("problem with event channel and dropped lock event");
        break;
      }
      if (eventChannel == null || !eventChannel.isConnected()) {
        LOG.debug("event channel is not open : can not send LockEvent message");
      } else {
        LOG.debug("LockEvent: " + eventMessage + " from cluster manager");
        if (eventMessage.getEventType() == LockService.Events.MASTER_FAILOVER.ordinal()) {
          eventChannel.send(null, null, eventMessage);
        } else {
          eventChannel.send(new Message(eventChannel.getView().getCreator(), null, eventMessage));
        }
        isMessageSent = true;
        break;
      }
      ++loopCount;
      sleeper.sleepMilliSeconds(100);
    }
  }

  private void sendLockEventToManager(EventMessage[] eventMessages) throws ChannelException {
    boolean isMessageSent = false;
    int loopCount = 0;
    while (!isMessageSent) {
      if (loopCount > 10) {
        LOG.debug("problem with event channel and dropped lock event");
        break;
      }
      if (eventChannel == null || !eventChannel.isConnected()) {
        LOG.debug("event channel is not open : can not send LockEvent message");
      } else {
        LOG.debug("LockEvent: " + eventMessages.length + " from cluster manager");
        eventChannel.send(new Message(eventChannel.getView().getCreator(), null, eventMessages));
        isMessageSent = true;
        break;
      }
      ++loopCount;
      sleeper.sleep();
    }
  }

  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
    if (protocol.equals(ClientSessionProtocol.class.getName())) {
      return ClientSessionProtocol.versionID;
    } else if (protocol.equals(LockServiceProtocol.class.getName())) {
      return LockServiceProtocol.versionID;
    } else if (protocol.equals(LockAdminProtocol.class.getName())) {
      return LockAdminProtocol.versionID;
    } else {
      throw new IOException("unknown protocol to cluster manager: " + protocol);
    }
  }

  public boolean createNode(LockMetaData metaData, boolean overWrite, boolean isDirectory) {
    LOG.debug("!!!createNode!!! " + metaData);
    updateHeartbeat(metaData.getCreator());

    boolean isSuccess = false;
    if (!isPrimaryNode || !isPrimaryPartition.get()) {
      LOG.fatal("not primary node or partition : " + metaData);
      return isSuccess;
    }
    if (!controller.isValidateLock(metaData.getLockId())) {
      LOG.fatal("invalid lock id : " + metaData);
      return isSuccess;
    }
    ClientLockInfo lockInfo = new ClientLockInfo(metaData.getCreator(), metaData.getLockId());
    lockInfo.setLockMode(metaData.getLockMode());
    if (!lock(lockInfo).equals("true")) {
      LOG.warn("createNode " + metaData.getCreator() + " does not have lock over "
          + metaData.getLockId());
      return isSuccess;
    }
    try {
      if (isDirectory) {
        if (createDir(metaData)) {
          isSuccess = true;
        } else {
          LOG.fatal("failed to create dir " + metaData);
        }
      } else {
        if (createParentDir(metaData)) {
          if (createFile(metaData, overWrite)) {
            isSuccess = true;
          } else {
            LOG.fatal("failed to create file " + metaData);
          }
        } else {
          LOG.fatal("failed to create parent dir " + metaData);
        }
      }
      if (isSuccess) {
        synchronized (lockEventQueue) {
          lockEventQueue.add(new EventMessage(metaData.getCreator(),
              LockServiceClient.Events.NODE_ADDED.ordinal(), metaData.getLockId(), isDirectory,
              metaData.getGenerationNumber()));
          lockEventQueue.notifyAll();
        }
      }
    } catch (Exception e) {
      LOG.error("error while creating node. meta : " + metaData, e);
      isSuccess = false;
    } finally {
      if (!isSuccess) {
        unlock(lockInfo);
      }
    }
    return isSuccess;
  }

  private boolean createFile(LockMetaData metaData, boolean overWrite) {
    boolean isSuccess = false;
    DataOutputStream out = null;
    boolean deleted = false;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      fileTxnManager.writeLock(txId, metaData.getLockId());
      if (fs.exists(new Path(conf.get("pleiades.root") + metaData.getLockId()))) {
        LOG.debug("file " + metaData.getLockId() + " exists");
        if (overWrite) {
          metaData.setGenerationNumber(System.nanoTime());
          if (!(deleted = fs.delete(new Path(conf.get("pleiades.root") + metaData.getLockId())))) {
            LOG.info("failed to delete file " + metaData.getLockId());
            fileTxnManager.releaseLock(txId, metaData.getLockId());
            return isSuccess;
          }
        } else {
          fileTxnManager.releaseLock(txId, metaData.getLockId());
          return isSuccess;
        }
        if (deleted) {
          LOG.debug("deleted file " + metaData.getLockId());
          synchronized (lockEventQueue) {
            lockEventQueue.add(new EventMessage(metaData.getCreator(),
                LockServiceClient.Events.NODE_REMOVED.ordinal(), metaData.getLockId(), false,
                metaData.getGenerationNumber()));
            lockEventQueue.notifyAll();
          }
        }
      }
      metaData.setGenerationNumber(System.nanoTime());
      out = new DataOutputStream(fs.create(new Path(conf.get("pleiades.root")
          + metaData.getLockId())));
      writeMetaData(metaData, out);
      out.flush();
      out.close();
      isSuccess = true;

    } catch (IOException e) {
      LOG.error("error while creating file. meta : " + metaData, e);
    } finally {
      fileTxnManager.releaseLock(txId, metaData.getLockId());
      if (out != null) {
        try {
          out.close();
        } catch (IOException e) {
        }
      }
    }
    return isSuccess;
  }

  private boolean createDir(LockMetaData metaData) {
    boolean isSuccess = false;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      fileTxnManager.writeLock(txId, metaData.getLockId());
      if (fs.exists(new Path(conf.get("pleiades.root") + metaData.getLockId()))) {
        LOG.fatal("path " + metaData.getLockId() + " already exists. failed to create directory");
      } else {
        Path path = new Path(conf.get("pleiades.root") + metaData.getLockId());
        if (!fs.mkdirs(path)) {
          if (fs.exists(path)) {
            LOG.fatal("");
          }
          fileTxnManager.releaseLock(txId, metaData.getLockId());
          return isSuccess;
        }
        isSuccess = true;
      }
    } catch (IOException e) {
      LOG.error("error while creating dir. meta : " + metaData, e);
    } finally {
      fileTxnManager.releaseLock(txId, metaData.getLockId());
    }
    return isSuccess;
  }

  private boolean createParentDir(LockMetaData metaData) {
    boolean isSuccess = false;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    String parentPath = "";
    try {
      if (metaData.getLockId().endsWith("/")) {
        parentPath = metaData.getLockId().substring(0, (metaData.getLockId().length() - 1));
      } else {
        parentPath = metaData.getLockId();
      }
      int index = parentPath.lastIndexOf("/");
      parentPath = parentPath.substring(0, index);
      fileTxnManager.writeLock(txId, parentPath);
      if (!fs.exists(new Path(conf.get("pleiades.root") + parentPath))) {
        if (!fs.mkdirs(new Path(conf.get("pleiades.root") + parentPath))) {
          LOG.warn("failed to mkdir " + conf.get("pleiades.root") + parentPath);
          fileTxnManager.releaseLock(txId, parentPath);
          return isSuccess;
        }
      }
      isSuccess = true;
    } catch (IOException e) {
      LOG.error("error while creating parent dir. meta : " + metaData, e);

    } finally {
      fileTxnManager.releaseLock(txId, parentPath);
    }

    return isSuccess;
  }

  public boolean deleteNode(ClientLockInfo lockInfo, boolean recursive) {
    LOG.debug("!!!deleteNode!!! " + lockInfo);
    updateHeartbeat(lockInfo.getCreator());

    boolean isSuccess = false;
    if (!isPrimaryNode || !isPrimaryPartition.get()) {
      LOG.fatal("not primary node or partition : " + lockInfo);
      return isSuccess;
    }
    boolean isNodeDirectory = false;
    long generationNumber = 0;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      if (!controller.isValidateLock(lockInfo.getLockId())) {
        LOG.debug("invalid lock id " + lockInfo);
        return isSuccess;
      }
      if (!lock(lockInfo).equals("true")) {
        LOG.warn("deleteNode " + lockInfo.getCreator() + " does not have lock over "
            + lockInfo.getLockId());
        return isSuccess;
      }

      boolean deleted = false;
      List<String> deletedNodes = new ArrayList<String>();
      fileTxnManager.writeLock(txId, lockInfo.getLockId());

      if (!fs.exists(new Path(conf.get("pleiades.root") + lockInfo.getLockId()))) {
        LOG.debug("deleteNode node " + lockInfo.getLockId() + " does not exist");
        fileTxnManager.releaseLock(txId, lockInfo.getLockId());
        return isSuccess;
      }
      if (fs.isDirectory(new Path(conf.get("pleiades.root") + lockInfo.getLockId()))) {
        isNodeDirectory = true;
      }
      if (recursive) {
        if (unlockAllUnderDir(lockInfo, deletedNodes)) {
          try {
            for (String entry : deletedNodes) {
              fileTxnManager.writeLock(txId, entry);
            }
            if ((deleted = fs.delete(new Path(conf.get("pleiades.root") + lockInfo.getLockId()),
                true))) {
              generationNumber = System.nanoTime();
              isSuccess = true;
            } else {
              LOG.warn("failed to delete recursive files " + lockInfo.getLockId()
                  + " and deleteNode failed");
            }
          } finally {
            for (String entry : deletedNodes) {
              fileTxnManager.releaseLock(txId, entry);
            }
          }
        } else {
          LOG.warn("unlockAllUnderdir failed " + lockInfo);
        }
      } else {
        if ((deleted = fs.delete(new Path(conf.get("pleiades.root") + lockInfo.getLockId())))) {
          generationNumber = System.nanoTime();
          isSuccess = true;
        } else {
          LOG.warn("failed to delete file " + lockInfo.getLockId() + " and deleteNode failed");
        }
      }

      fileTxnManager.releaseLock(txId, lockInfo.getLockId());

      if (deleted) {
        distributedLockStore.remove(lockInfo.getLockId());
      }

      if (isSuccess) {
        try {
          EventMessage eventMessage = new EventMessage(lockInfo.getCreator(),
              LockServiceClient.Events.NODE_REMOVED.ordinal(), lockInfo.getLockId(),
              isNodeDirectory, generationNumber);
          if (!isNodeDirectory) {
            for (String entry : deletedNodes) {
              eventMessage.setLockId(entry);
              eventMessage.setIsDirectory(false);
              sendLockEventToManager(eventMessage);
            }
          }
          synchronized (lockEventQueue) {
            eventMessage.setLockId(lockInfo.getLockId());
            eventMessage.setIsDirectory(isNodeDirectory);
            lockEventQueue.add(eventMessage);
            lockEventQueue.notifyAll();
          }
        } catch (Exception e) {
          // sending event failure does not affect the lock operation result ...
          LOG.error("error while processing event when deleting node. lockInfo : " + lockInfo, e);
        }
      }
    } catch (IOException e) {
      LOG.error("error while deleting node. lockInfo : " + lockInfo, e);
      isSuccess = false;

    } finally {
      fileTxnManager.releaseLock(txId, lockInfo.getLockId());
      if (!isSuccess) {
        unlock(lockInfo);
      }
    }
    return isSuccess;
  }

  private boolean unlockAllUnderDir(ClientLockInfo lockInfo, List<String> deletedNodes) {
    boolean isSuccess = false;
    LockMetaData[] metaData = null;
    try {
      metaData = readDirWithFSLock(lockInfo.getLockId(), true, false);
      if (metaData != null) {
        for (LockMetaData entry : metaData) {
          ClientLockInfo childNodeInfo = new ClientLockInfo(lockInfo.getCreator(), entry
              .getLockId());
          childNodeInfo.setLockMode(LockServiceClient.LockMode.LOCK_EXCLUSIVE.ordinal());
          deletedNodes.add(entry.getLockId());
          if (!unlock(childNodeInfo).equals("true")) {
            deletedNodes.clear();
            LOG.warn("node " + lockInfo.getLockId() + " already locked by other host : "
                + entry.getCreator());
            return isSuccess;
          }
        }
      }
      isSuccess = true;
    } catch (Exception e) {
      LOG.error("error while unlocking all under dir, lockinfo : " + lockInfo, e);
    }

    return isSuccess;
  }

  public String getContents(ClientLockInfo lockInfo) {
    updateHeartbeat(lockInfo.getCreator());

    DataInputStream in = null;
    String contents = "";
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      fileTxnManager.readLock(txId, lockInfo.getLockId());
      in = fs.openDataInputStream(new Path(conf.get("pleiades.root") + lockInfo.getLockId()));
      getMetaData(in);
      contents = WritableUtils.readString(in);
      in.close();
    } catch (IOException e) {
      e.printStackTrace();
    } finally {
      fileTxnManager.releaseLock(txId, lockInfo.getLockId());
      if (in != null)
        try {
          in.close();
        } catch (IOException e) {
          LOG.error("error while closing file", e);
        }
    }
    return contents;
  }

  public boolean setContents(ClientLockInfo lockInfo, String contents) {
    updateHeartbeat(lockInfo.getCreator());

    DataOutputStream out = null;
    boolean isSuccess = false;
    if (!isPrimaryNode || !isPrimaryPartition.get()) {
      LOG.fatal("not primary node or partition : " + lockInfo);
      return isSuccess;
    }
    if (!controller.isValidateLock(lockInfo.getLockId())) {
      LOG.debug("invalid lock id " + lockInfo);
      return isSuccess;
    }
    if (!lock(lockInfo).equals("true")) {
      LOG.debug("setContents " + lockInfo.getCreator() + " does not have lock over "
          + lockInfo.getLockId());
      return isSuccess;
    }

    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      LockMetaData metaData = null;
      Path path = new Path(conf.get("pleiades.root") + lockInfo.getLockId());
      boolean deleted = false;
      fileTxnManager.writeLock(txId, lockInfo.getLockId());

      if (fs.exists(path)) {
        metaData = getMetaDataWithFSLock(lockInfo.getLockId(), false);
        metaData.setGenerationNumber(System.nanoTime());
        deleted = fs.delete(path);
      }
      if (metaData == null) {
        metaData = new LockMetaData(lockInfo.getLockId());
      }
      metaData.setCreator(lockInfo.getCreator());
      metaData.setLockMode(lockInfo.getLockMode());
      metaData.setGenerationNumber(System.nanoTime());

      out = new DataOutputStream(fs.create(path));
      writeMetaData(metaData, out);
      WritableUtils.writeString(out, contents);
      out.flush();
      out.close();

      fileTxnManager.releaseLock(txId, lockInfo.getLockId());

      if (deleted) {
        synchronized (lockEventQueue) {
          lockEventQueue.add(new EventMessage(metaData.getCreator(),
              LockServiceClient.Events.NODE_MODIFIED.ordinal(), metaData.getLockId(), false,
              metaData.getGenerationNumber()));
          lockEventQueue.notifyAll();
        }
      }
      synchronized (lockEventQueue) {
        lockEventQueue.add(new EventMessage(metaData.getCreator(),
            LockServiceClient.Events.CONTENTS_CHANGED.ordinal(), metaData.getLockId(), false,
            (metaData.getGenerationNumber() + 1)));
        lockEventQueue.notifyAll();
      }

      isSuccess = true;

    } catch (IOException e) {
      LOG.error("error while setting content. lockInfo : " + lockInfo, e);
    } finally {
      fileTxnManager.releaseLock(txId, lockInfo.getLockId());
      if (out != null)
        try {
          out.flush();
          out.close();
        } catch (IOException e) {
          LOG.error("error while closing file in setContents()", e);
        }
      if (!isSuccess) {
        unlock(lockInfo);
      }
    }
    return isSuccess;
  }

  private void writeMetaData(LockMetaData metaData, DataOutputStream out) throws IOException {
    WritableUtils.writeString(out, metaData.getCreator());
    WritableUtils.writeString(out, metaData.getLockId());
    out.writeInt(metaData.getLockMode());
    out.writeLong(metaData.getGenerationNumber());
    WritableUtils.writeString(out, metaData.getContentKey());
  }

  public boolean nodeExist(String path) throws IOException {
    try {
      String metaPath = conf.get("pleiades.root") + path;
      Path lockPath = new Path(metaPath);
      if (fs.exists(lockPath)) {
        LOG.debug("nodeExist : path ");
        return true;
      }
    } catch (Exception e) {
      LOG.error("error while checking node existence path : " + path, e);
      throw new IOException(e.getMessage());
    }

    return false;
  }

  public LockMetaData getMetaData(String path) {
    return getMetaDataWithFSLock(path, true);
  }

  private LockMetaData getMetaDataWithFSLock(String path, boolean lock) {
    DataInputStream in = null;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      String metaPath = conf.get("pleiades.root") + path;
      Path lockPath = new Path(metaPath);
      if (lock) {
        fileTxnManager.readLock(txId, path);
      }
      if (!fs.exists(lockPath)) {
        LOG.debug("getMetaData : path " + path + " does not exist");
        return null;
      }
      LockMetaData metaData = new LockMetaData(path);
      if (!fs.isDirectory(lockPath)) {
        in = fs.openDataInputStream(lockPath);
        metaData = getMetaData(in);
      }
      return metaData;
    } catch (IOException e) {
      LOG.error("error while getting metadata with fs lock. path : " + path, e);
    } finally {
      if (lock) {
        fileTxnManager.releaseLock(txId, path);
      }
      if (in != null)
        try {
          in.close();
        } catch (IOException e) {
          LOG.error("error while closing file in getMetaDataWithFSLock()", e);
        }
    }
    return null;
  }

  public LockMetaData[] readDir(String path, boolean includeDirectory) {
    return readDirWithFSLock(path, includeDirectory, true);
  }

  private LockMetaData[] readDirWithFSLock(String path, boolean includeDirectory,
      boolean lockRootPath) {

    DataInputStream in = null;
    List<LockMetaData> metaDataList = new ArrayList<LockMetaData>();
    LockMetaData[] metaDatas = null;
    String txId = String.valueOf(fileTxnNumber.getAndIncrement());
    try {
      String rootPath = conf.get("pleiades.root");
      String metaPath = rootPath + path;
      if (lockRootPath) {
        fileTxnManager.readLock(txId, path);
      }
      if (!fs.exists(new Path(metaPath))) {
        LOG.debug("readDir : path " + path + " does not exist");
        return metaDatas;
      }
      Path[] paths = null;
      paths = fs.list(new Path(metaPath));
      if (paths == null || paths.length <= 0) {
        LOG.debug("readDir : path " + path + " does not have any entry");
        return metaDatas;
      }
      try {
        for (Path entry : paths) {
          String strEntry = entry.toString();
          if (rootPath != null && strEntry.startsWith(rootPath)) {
            strEntry = strEntry.substring(rootPath.length());
          }
          fileTxnManager.readLock(txId, strEntry);
        }
        for (Path entry : paths) {
          LockMetaData metaData;
          metaData = new LockMetaData(path + "/" + entry.getName());
          if (fs.isDirectory(entry)) {
            metaDataList.add(metaData);
            if (includeDirectory) {
              LockMetaData[] subMetadatas = null;
              subMetadatas = readDir(path + "/" + entry.getName(), true);
              if (subMetadatas != null) {
                for (LockMetaData subEntry : subMetadatas) {
                  metaDataList.add(subEntry);
                }
              }
            }
          } else {
            in = fs.openDataInputStream(entry);
            metaData = getMetaData(in);
            metaDataList.add(metaData);
            in.close();
          }
        }
      } finally {
        for (Path entry : paths) {
          String strEntry = entry.toString();
          if (rootPath != null && strEntry.startsWith(rootPath)) {
            strEntry = strEntry.substring(rootPath.length());
          }
          fileTxnManager.releaseLock(txId, strEntry);
        }
      }
      if (metaDataList.size() > 0) {
        metaDatas = new LockMetaData[metaDataList.size()];
        metaDataList.toArray(metaDatas);
      }

    } catch (IOException e) {
      LOG.error("error in readDirWithFSLock", e);
    } finally {
      if (lockRootPath) {
        fileTxnManager.releaseLock(txId, path);
      }
      if (in != null)
        try {
          in.close();
        } catch (IOException e) {
          LOG.error("error while closing file", e);
        }
    }
    return metaDatas;
  }

  private LockMetaData getMetaData(DataInputStream in) throws IOException {
    LockMetaData metaData = new LockMetaData();
    metaData.setCreator(WritableUtils.readString(in));
    metaData.setLockId(WritableUtils.readString(in));
    metaData.setLockMode(in.readInt());
    metaData.setGenerationNumber(in.readLong());
    metaData.setContentKey(WritableUtils.readString(in));
    return metaData;
  }

  public String getMasterPleiades() {
    return masterPleiades;
  }

  public String[] getSlavePleiades() {
    String[] slaves = null;
    synchronized (slaveManagersQueue) {
      if (slavePleiadesList.isEmpty()) {
        slaves = new String[1];
        slaves[0] = getHostName();
      } else {
        slaves = new String[slavePleiadesList.size()];
        slavePleiadesList.toArray(slaves);
      }
    }
    return slaves;
  }

  public String getMasterEventManager() {
    return masterEventManager;
  }

  public String[] getSlaveEventManager() {
    String[] slaveEventManagers = null;
    synchronized (slaveEventManagerList) {
      slaveEventManagers = new String[slaveEventManagerList.size()];
      int index = 0;
      for (String eventManager : slaveEventManagerList) {
        slaveEventManagers[index] = eventManager;
        ++index;
      }
    }
    return slaveEventManagers;
  }

  public String getServiceId(String serviceName) {
    boolean isSuccess = false;
    try {
      String clientIp = New_Server.getRemoteAddress();
      if (clientIp == null) {
        clientIp = "";
      }
      LOG.debug("service name : " + serviceName + ", client ip : " + clientIp);
      String serviceId = null;
      if ((serviceId = controller.getPathForService(clientIp, serviceName)) != null) {
        isSuccess = true;
        return serviceId;
      }
    } catch (IOException e) {
      LOG.error("error while retrieving service id. serviceName : " + serviceName, e);
    } finally {
    }
    return String.valueOf(isSuccess);
  }

  public String getLockDump() {
    StringBuffer lockStatus = new StringBuffer();
    for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
      lockStatus.append(hostName + " lock status : " + entry.getKey() + ", " + entry.getValue()
          + "\n");
    }
    return lockStatus.toString();
  }

  public LockStatus getLockStatus(String path) {
    Integer memTxn = memoryTxnNumber.getAndIncrement();
    try {
      memoryTxnManager.readLock(memTxn, path);
      LockStatus lockStatus = distributedLockStore.get(path);
      LOG.warn("Path:" + path + ":" + lockStatus);
      return lockStatus;
    } finally {
      memoryTxnManager.release(memTxn, path);
    }

  }

  public boolean isPrimaryMaster() {
    if (isPrimaryPartition.get()) {
      return isPrimaryNode;
    } else
      return false;
  }

  public PleiadesInfoData getPleiadesInfo() {
    PleiadesInfoData data = new PleiadesInfoData();
    data.setClusterId(this.clusterID);
    data.setNotificationChannel(getNotificationChannel());
    data.setDfsRoot(getDfsRoot());
    data.setSplitBrainMode(isSplitBrainMode());
    data.setHeartbeatPort(conf.getInt("clustermanager.heartbeat.port", 11000));
    data.setEventUsingMulticast(conf.getBoolean("event.delivery.multicast", true));
    return data;
  }

  private void updateHeartbeat(String creator) {
    LOG.debug("creator : " + creator + " updates heartbeat");
    heartBeatMap.put(creator, new Long(System.currentTimeMillis()));
    timeSinceLastGC.incrementAndGet();
  }

  // ///////////////////////////////////////////////////
  // ClientSession Protocol
  // ///////////////////////////////////////////////////

  public void heartbeat(ClientSessionInfo clientInfo) throws IOException {
    // try {
    // LOG.debug("heartbeat " + clientInfo);
    // distributedSessionStore.put(clientInfo.getHostName(), new
    // Long(System.currentTimeMillis()));
    // } catch (Exception e) {
    // e.printStackTrace(System.out);
    // throw new IOException(e.getMessage());
    // }
  }

  public boolean isPrimaryPartition() throws IOException {
    return isPrimaryPartition.get();
  }

  public boolean checkServer() {
    return true;
  }

  // ///////////////////////////////////////////////////
  // ClusterManagerInterface Other Public Methods
  // ///////////////////////////////////////////////////

  public void dumpLockStatus() {
    LOG.info("+++++++++++++++ lock status dump starts +++++++++++++++");
    for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
      LOG.info(hostName + " lock status : " + entry.getKey() + ", " + entry.getValue());
    }
    LOG.info("+++++++++++++++ lock status dump ends +++++++++++++++");
  }

  public String getHostName() {
    return this.hostName;
  }

  public JChannel getClusterChannel() {
    return clusterChannel;
  }

  public void setClusterChannel(JChannel clusterChannel) {
    if (clusterChannel != null) {
      clusterChannel.close();
    }
    this.clusterChannel = clusterChannel;
  }

  public void shutdown() {
    this.stopRequested.set(true);
    clusterChannelThread.interrupt();
    eventChannelThread.interrupt();
    heartBeatMonitorThread.interrupt();
    lockServiceThread.interrupt();
    invalidLockList.clear();
    for (LockEventSender sender : eventSenderList) {
      sender.interrupt();
    }
    sleeper.sleep();
    server.stop();
    try {
      ((HeartbeatXceiveServer) this.heartbeatXceiveServer.getRunnable()).kill();
    } catch (Exception e) {
      LOG.error(e);
    }
  }

  // ///////////////////////////////////////////////////
  // ClusterManager Internal Methods
  // ///////////////////////////////////////////////////

  private boolean isSplitBrainMode() {
    return conf.getBoolean("pleiades.mode.splitBrain", false);
  }

  private String getDfsRoot() {
    return this.dfsRoot;
  }

  private String getNotificationChannel() {
    return (clusterID + conf.get("notificationChannelName",
        "com.nhncorp.neptune.notificationChannel"));
  }

  private void startupEventSender() throws Throwable {
    lockEventQueue = new LinkedList<EventMessage>();
    int handlerCount = conf.getInt("pleiades.lockEventHandler.count", 3);
    eventSenderList = new ArrayList<LockEventSender>();
    for (int count = 0; count < handlerCount; ++count) {
      LockEventSender eventSender = new LockEventSender("event sender thread "
          + String.valueOf(count));
      eventSender.start();
      eventSenderList.add(eventSender);
    }
  }

  class LockEventSender extends Thread {
    public LockEventSender(String name) {
      super(name);
    }

    public void run() {
      while (!stopRequested.get()) {
        try {
          EventMessage[] events = null;
          synchronized (lockEventQueue) {
            while (lockEventQueue.isEmpty()) {
              lockEventQueue.wait();
            }
            events = lockEventQueue.toArray(new EventMessage[] {});
            lockEventQueue.clear();
          }
          sendLockEventToManager(events);
          events = null;

        } catch (InterruptedException e) {
          return;
        } catch (ChannelException e) {
          LOG.error("error while sending lock event", e);
        }
      }
    }
  }

  // ///////////////////////////////////////////////////
  // LockAdmin Protocol
  // ///////////////////////////////////////////////////

  public boolean admin_create(AdminLockInfo lockInfo, boolean overwrite) {
    LOG.debug("admin_create " + lockInfo);
    boolean isSuccess = false;
    if (admin_lock(lockInfo)) {
      String lockId = getFullPath(lockInfo);
      try {
        LockMetaData metaData = new LockMetaData(lockId);
        metaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
        metaData.setCreator(lockInfo.getOwner());
        if (!createParentDir(metaData)) {
          LOG.fatal("failed to create parent dir " + metaData);
          return isSuccess;
        }
        if (createFile(metaData, overwrite)) {
          isSuccess = true;
        } else {
          LOG.fatal("failed to create file " + metaData);
        }
      } catch (Exception e) {
        e.printStackTrace();
        isSuccess = false;
      } finally {
        if (!isSuccess) {
          admin_release(lockInfo);
        }
      }
    }
    return isSuccess;
  }

  public boolean admin_delete(AdminLockInfo lockInfo, boolean recursive) {
    LOG.debug("admin_delete " + lockInfo);
    boolean isSuccess = false;
    try {
      String lockId = getFullPath(lockInfo);
      if (!fs.exists(new Path(conf.get("pleiades.root") + lockId))) {
        LOG.debug("deleteNode " + lockInfo.getLockId() + " node does not exist");
        return isSuccess;
      }
      if (recursive) {
        if (!admin_release_recursive(lockInfo)) {
          LOG.warn("failed to release recursive files " + lockInfo.getLockId()
              + " and deleteNode failed");
          return isSuccess;
        }
        if (fs.delete(new Path(conf.get("pleiades.root") + lockId), true)) {
          isSuccess = true;
        } else {
          LOG.warn("failed to delete recursive files " + lockInfo.getLockId()
              + " and deleteNode failed");
        }
      } else {
        if (!admin_release(lockInfo)) {
          LOG.warn("failed to release files " + lockInfo.getLockId() + " and deleteNode failed");
          return isSuccess;
        }
        if (fs.delete(new Path(conf.get("pleiades.root") + lockId))) {
          isSuccess = true;
        } else {
          LOG.warn("failed to delete file " + lockInfo.getLockId() + " and deleteNode failed");
        }
      }
    } catch (IOException e) {
      e.printStackTrace();
      isSuccess = false;
    } finally {
    }
    return isSuccess;
  }

  public boolean admin_set(AdminLockInfo lockInfo, String keyword) {
    LOG.debug("admin_set " + lockInfo + ", " + keyword);
    boolean isSuccess = false;
    if (admin_lock(lockInfo)) {
      String lockId = getFullPath(lockInfo);
      try {
        LockMetaData metaData = new LockMetaData(lockId);
        metaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());
        metaData.setCreator(lockInfo.getOwner());
        metaData.setContentKey(keyword);
        if (!createParentDir(metaData)) {
          LOG.fatal("failed to create parent dir " + metaData);
          return isSuccess;
        }
        if (createFile(metaData, true)) {
          isSuccess = true;
        } else {
          LOG.fatal("failed to create file " + metaData);
        }
      } catch (Exception e) {
        e.printStackTrace();
        isSuccess = false;
      } finally {
        if (!isSuccess) {
          admin_release(lockInfo);
        }
      }
    }
    return isSuccess;
  }

  public LockMetaData admin_getMeta(AdminLockInfo lockInfo) {
    LOG.debug("admin_getMeta " + lockInfo);
    String lockId = getFullPath(lockInfo);
    return getMetaData(lockId);
  }

  public LockMetaData[] admin_getMetaList(AdminLockInfo lockInfo) {
    LOG.debug("admin_getMetaList " + lockInfo);
    String lockId = getFullPath(lockInfo);
    return readDir(lockId, true);
  }

  public LockMetaData[] admin_listAll(AdminLockInfo lockInfo) {

    LockMetaData[] metaDatas = null;
    List<LockMetaData> metaList = new ArrayList<LockMetaData>();

    for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
      if (!lockInfo.getServiceId().equals("")) {
        if (!entry.getKey().startsWith(getFullPath(lockInfo))) {
          continue;
        }
      }
      LockMetaData metaData = new LockMetaData(entry.getValue().getLockId());
      metaData.setCreator(entry.getValue().getHostName());
      metaData.setGenerationNumber(entry.getValue().getLastCommitTime());
      metaData.setLockMode(entry.getValue().getMode());
      metaList.add(metaData);
    }

    if (!metaList.isEmpty()) {
      metaDatas = metaList.toArray(new LockMetaData[metaList.size()]);
    }

    return metaDatas;
  }

  public LockMetaData[] admin_listById(AdminLockInfo lockInfo) {
    LockMetaData[] metaDatas = null;
    List<LockMetaData> metaList = new ArrayList<LockMetaData>();
    int index = 0;
    String servicePath = getServiceId(lockInfo.getServiceId());
    for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
      if (entry.getKey().startsWith(servicePath) && entry.getKey().contains(lockInfo.getLockId())) {
        LockMetaData metaData = new LockMetaData(entry.getValue().getLockId());
        metaData.setCreator(entry.getValue().getHostName());
        metaData.setGenerationNumber(entry.getValue().getLastCommitTime());
        metaList.add(metaData);
      }
      ++index;
    }
    if (!metaList.isEmpty()) {
      metaDatas = metaList.toArray(new LockMetaData[metaList.size()]);
    }

    return metaDatas;
  }

  public LockMetaData[] admin_listByOwner(AdminLockInfo lockInfo) {
    LockMetaData[] metaDatas = null;
    List<LockMetaData> metaList = new ArrayList<LockMetaData>();
    int index = 0;
    String servicePath = getServiceId(lockInfo.getServiceId());
    for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
      if (lockInfo.getOwner().equals(entry.getValue().getHostName())
          && entry.getKey().startsWith(servicePath)) {
        LockMetaData metaData = new LockMetaData(entry.getValue().getLockId());
        metaData.setCreator(entry.getValue().getHostName());
        metaData.setGenerationNumber(entry.getValue().getLastCommitTime());
        metaList.add(metaData);
      }
      ++index;
    }
    if (!metaList.isEmpty()) {
      metaDatas = metaList.toArray(new LockMetaData[metaList.size()]);
    }

    return metaDatas;
  }

  public String[] admin_listPath(AdminLockInfo lockInfo) {
    String lockId = getFullPath(lockInfo);
    String resultPaths[] = null;
    Path[] paths = null;
    try {
      paths = fs.list(new Path(conf.get("pleiades.root") + lockId));
      if (paths != null && paths.length > 0) {
        resultPaths = new String[paths.length];
        int index = 0;
        for (Path entry : paths) {
          resultPaths[index] = entry.toString();
          ++index;
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
    }

    return resultPaths;
  }

  public boolean admin_lock(AdminLockInfo lockInfo) {
    String lockId = getFullPath(lockInfo);
    LockStatus lockStatus = new LockStatus(lockInfo.getOwner(), lockInfo.getLockId(),
        LockServiceClient.LockMode.LOCK_EXCLUSIVE.ordinal());
    Integer memTxn = memoryTxnNumber.getAndIncrement();
    try {
      memoryTxnManager.writeLock(memTxn, lockId);
      distributedLockStore.put(lockId, lockStatus);
    } finally {
      memoryTxnManager.release(memTxn, lockId);
    }
    return true;
  }

  public boolean admin_release(AdminLockInfo lockInfo) {
    String lockId = getFullPath(lockInfo);
    Integer memTxn = memoryTxnNumber.getAndIncrement();
    try {
      memoryTxnManager.writeLock(memTxn, lockId);
      distributedLockStore.remove(lockId);
    } finally {
      memoryTxnManager.release(memTxn, lockId);
    }
    return true;
  }

  private boolean admin_release_recursive(AdminLockInfo lockInfo) throws IOException {
    String lockId = getFullPath(lockInfo);
    Path[] paths = fs.list(new Path(conf.get("pleiades.root") + lockId));
    if (paths != null && paths.length > 0) {
      for (Path entry : paths) {
        Integer memTxn = memoryTxnNumber.getAndIncrement();
        try {
          memoryTxnManager.writeLock(memTxn, entry.toString());
          distributedLockStore.remove(entry.toString());
        } finally {
          memoryTxnManager.release(memTxn, entry.toString());
        }
      }
    }
    admin_release(lockInfo);
    return true;
  }

  public boolean admin_releaseByOwner(String owner) {
    boolean isSuccess = false;
    try {
      for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
        if (owner.equals(entry.getValue().getHostName())) {
          Integer memTxn = memoryTxnNumber.getAndIncrement();
          try {
            memoryTxnManager.writeLock(memTxn, entry.getKey());
            LOG.debug("removing lock :" + entry.getKey());
            distributedLockStore.remove(entry.getKey());
          } finally {
            memoryTxnManager.release(memTxn, entry.getKey());
          }
        }
      }
      isSuccess = true;
    } catch (Exception e) {
      e.printStackTrace();
      isSuccess = false;
    } finally {
    }

    return isSuccess;
  }

  public boolean admin_releaseByService(String serviceId) {
    String fullServiceId = getServiceId(serviceId);
    boolean isSuccess = false;
    try {
      for (Map.Entry<String, LockStatus> entry : distributedLockStore.entrySet()) {
        if (entry.getKey().startsWith(fullServiceId)) {
          Integer memTxn = memoryTxnNumber.getAndIncrement();
          try {
            memoryTxnManager.writeLock(memTxn, entry.getKey());
            LOG.debug("removing lock :" + entry.getKey());
            distributedLockStore.remove(entry.getKey());
          } finally {
            memoryTxnManager.release(memTxn, entry.getKey());
          }
        }
      }
      isSuccess = true;
    } catch (Exception e) {
      e.printStackTrace();
      isSuccess = false;
    } finally {
    }

    return isSuccess;
  }

  public boolean admin_releaseAll() {
    distributedLockStore.clear();
    return true;
  }

  public boolean admin_format() {
    boolean isSuccess = true;
    distributedLockStore.clear();
    Path[] paths = null;
    try {
      paths = fs.list(new Path(conf.get("pleiades.root")));
      if (paths != null && paths.length > 0) {
        for (Path entry : paths) {
          if (!fs.delete(entry, true)) {
            isSuccess = false;
            LOG.fatal("failed to delete " + entry.toString() + " and format failed");
            break;
          }
        }
      }
    } catch (Exception e) {
      isSuccess = false;
      e.printStackTrace();
    }
    return isSuccess;
  }
  
  public boolean isPrimaryPleiades() {
    if (isPrimaryPartition.get()) {
      return isPrimaryNode;
    } else
      return false;
  }

  private String getFullPath(AdminLockInfo lockInfo) {
    StringBuffer fullPath = new StringBuffer("");
    fullPath.append(getServiceId(lockInfo.getServiceId()));
    if (!lockInfo.getLockId().startsWith("/")) {
      fullPath.append("/");
    }
    if (lockInfo.getLockId().contains("..")) {
      fullPath.append("/");
    } else {
      fullPath.append(lockInfo.getLockId());
    }
    return fullPath.toString();
  }

  // ///////////////////////////////////////////////////
  // Main Thread Run
  // ///////////////////////////////////////////////////

  public void run() {
    try {
      startupEventSender();
      this.server.join();
    } catch (Throwable e) {
      if (e instanceof InterruptedException) {
        return;
      } else {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) throws Exception {

    ClusterManager clusterManager = new ClusterManager("conf/session.prop", null);
    new Thread(clusterManager, "main cluster manager thread").start();
  }

}
